package job.gfp.ranks.calc;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;

import job.gfp.ranks.base.Data;
import job.gfp.ranks.readable.Ranks_Readable;
import mapred.SimpleJob;
import mapred.config.Compress;
import mapred.config.JobConfig;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.RunningJob;

@SuppressWarnings("unchecked")
public class Calc extends SimpleJob {

    private int round;

    private String output;
    
    private String[] settings;
    
    public Calc(int round, String... settings) throws IOException {
        super("Ranks Calc Round " + round, settings);
        
        this.round = round;
        this.output = this.getUserSetting("-output");
        this.settings = settings;
    }

    @Override
    protected Class<? extends Reducer> getCombinerClass() {
        return Calc_Combiner.class;
    }

    @Override
    protected Compress getCompress() {
        return Compress.GZIP;
    }

    @Override
    protected String[] getInputPaths(FileSystem fs) throws IOException {
        return new String[] { this.getUserSetting("-output") + "/Round_"
                + (round - 1) };
    }

    @Override
    protected Class<? extends Mapper> getMapperClass() {
        return Calc_Mapper.class;
    }

    @Override
    protected String getOutputPath(FileSystem fs) throws IOException {
        return this.getUserSetting("-output") + "/Round_" + round;
    }

    @Override
    protected Class<? extends Reducer> getReducerClass() {
        return Calc_Reducer.class;
    }

    @Override
    protected int getReducerNumber() {
        return 1000;
    }

    @Override
    protected void setUp(JobConfig conf, FileSystem fs) throws IOException {
        conf.set("Output", output);
        conf.set("Round", round);

        long websize = Long.parseLong(new BufferedReader(new InputStreamReader(
                fs.open(new Path(output + "/WebSize")))).readLine());
        conf.set("WebSize", websize);

        String z = "";
        if (round == 1) {
            z = new Data().toString();
        } else {
            try {
                z = new BufferedReader(new InputStreamReader(fs.open(new Path(
                        output + "/Z_" + (round - 1))))).readLine();
            } catch (IOException e) {
                z = new Data().toString();
            }
        }
        conf.set("Z", z);
    }

    @Override
    protected void tearDown(JobConfig conf, FileSystem fs, RunningJob job)
            throws IOException {
        long delta = this.getCounterSumByGroup(job, "Delta");
        long distance = this.getCounterSumByGroup(job, "Distance");

        if (distance <= delta) {
            try {
                new Ranks_Readable(round, settings).submit();
            } catch (Exception e) {
                throw new IOException(e);
            }
        }

        fs.delete(new Path(output + "/Round_" + (round - 1)), true);
        fs.delete(new Path(output + "/Z_" + (round - 1)), true);
    }

}
